package com.pfdu.simplescheduler.schedule;

import com.pfdu.simplescheduler.common.Task;
import com.pfdu.simplescheduler.common.TaskInvoker;
import com.pfdu.simplescheduler.common.TaskStatus;
import com.pfdu.simplescheduler.config.SimpleSchedulerConfig;
import com.pfdu.simplescheduler.repo.TaskRepo;
import com.pfdu.simplescheduler.strategy.FilterContext;
import com.pfdu.simplescheduler.strategy.FilterStrategy;
import com.pfdu.simplescheduler.util.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.*;

@Slf4j
@Component
public class TaskScheduler {
    /**
     * 应用关闭信号
     */
    private volatile Boolean shutdown = false;

    private ThreadPoolExecutor workerGroup; // 性能调优项，可配置，根据应用服务器的CPU核心数设置，CPU core * 2 即可。

    @Autowired
    private TaskRepo taskRepo;

    @Autowired
    private SimpleSchedulerConfig config;

    @Autowired
    FilterContext filterContext;

    //    @PostConstruct
    public void init() {
        workerGroup = new ThreadPoolExecutor(config.getExecutorSize(), config.getExecutorSize(), 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(config.getQueueSize()), new ThreadPoolExecutor.AbortPolicy());
        new Thread(new Loader(filterContext.choose(config.getStrategy()))).start();
    }

    /**
     * Loader 任务加载线程
     * 1.延迟启动，即启动时先睡眠5s，等Spring ApplicationContext加载完毕后再运行。
     * 2.开始运行，根据任务筛选策略，决定应用节点信息是否影响任务筛选，默认策略为乐观锁抢任务，这时就忽略应用节点信息。
     * 3.批量读取任务。
     * 4.根据任务筛选策略去筛选任务（默认策略为多个应用实例通过乐观锁机制抢占任务）。
     * 5.锁定任务，将筛选出的任务在DB中的状态改为“待执行”。
     * 6.锁定成功的话，就将任务提交给工作组 WorkerGroup。如果没有空闲的 Worker，WorkerGroup 会抛出“拒绝执行的异常”。
     * 7.一旦 WorkerGroup 拒绝执行，就休息一定时间后再向 WorkerGroup 提交任务。
     * n.周期性睡眠，然后循环运行。
     *
     * 备注：
     * 第7步有2种处理方案：
     * 一种是现行的这种处理方式，缺点是代码可读性稍差。
     * 另一种是提交任务给线程池被拒绝后，将Task的status恢复为初始状态（未开始），代码可读性强，但缺点是会加重DB的IO。
     */
    class Loader implements Runnable {
        /**
         * 任务筛选器
         */
        private FilterStrategy filter;

        private Loader(FilterStrategy filter) {
            this.filter = filter;
        }

        @Override
        public void run() {
            try {
                Thread.sleep(5 * 1000);
            } catch (InterruptedException e) {
                log.warn("TaskLoader is interrupted when sleep after started.");
            }

            log.info("TaskLoader is started.");

            while (!shutdown) {
                try {
                    List<Task> tasks = taskRepo.listNotStarted();
                    Optional.ofNullable(tasks).orElse(new ArrayList<>(0))
                            .forEach((task) -> {
                                if (filter.accept(task)) {
                                    task.setStatus(TaskStatus.PENDING);
                                    task.setUpdateTime(LocalDateTime.now());
                                    int updateResult = taskRepo.updateWithVersion(task);
                                    if (updateResult > 0) {
                                        try {
                                            /**
                                             * 这里再查一次是因为上面乐观锁已经更新了版本，task对象的version值已经不是最新的了，需要重新查询。
                                             */
                                            task = taskRepo.get(task.getId());
                                            // 指派任务给workerGroup
                                            allotTask(task);
                                            log.debug(String.format("TaskLoader lock Task[%s] successfully.", task.getId()));
                                        } catch (RejectedExecutionException e) {
                                            log.info(String.format("There is no idle worker to execute Task[id=%s] in WorkerGroup, so WorkerGroup rejected execution.If there is a large number print of this log, consider turn up <simple-scheduler.task-load-interval>.", task.getId()));
                                        }
                                    }
                                }
                            });
                } catch (DataAccessResourceFailureException e) {
                    log.error("TaskLoader can not load or lock task in DB, TaskLoader may can not connect DB :", e);
                } catch (Exception e) {
                    log.error("TaskLoader is error:", e);
                }

                log.debug("TaskLoader load Task over.");

                try {
                    Thread.sleep(config.getLoadInterval()); // 性能调节项，可配置
                } catch (InterruptedException e) {
                    log.warn("TaskLoader is interrupted in periodic sleep.");
                }
            }
            log.info("TaskLoader is stopped.");
        }

        private void allotTask(Task task){
            // 当WorkGroup中没有空闲的Worker（线程）时，workerGroup会拒绝execute()，这时会抛出拒绝执行异常。
            boolean success = false;
            do {
                try {
                    if (task != null)
                        workerGroup.execute(new Work(task, taskRepo));
                    success = true;
                } catch (RejectedExecutionException e) {
                    log.warn("There is no idle worker in WorkerGroup, so WorkerGroup rejected execution.");
                    try {
                        Thread.sleep(config.getBossAllotTaskInterval()); // 性能调优项，可配置，Boss向WorkerGroup提交Task的速率，只有在任务执行时间较长时该值才会生效。
                    } catch (InterruptedException e1) {
                        log.error("TaskBoss is interrupted while sleeping after submitting task into WorkerGroup.");
                    }
                }
            } while (!success && !shutdown && !workerGroup.isShutdown());
        }
    }

    /**
     * Work 工作线程的工作内容
     * 1.检查任务是否有效，如果任务已被其它应用执行或已被重置（库中的status值已经改变）或已被其它应用锁定（version值已经落后于库中的值），就不执行该任务。
     * 2.将任务状态更新为“执行中”。
     * 3.执行任务。
     * 4.执行完任务，更新任务状态为“已完成”或者“执行异常”。
     */
    class Work implements Runnable {

        private Task task;
        private TaskRepo taskRepo;

        Work(Task task, TaskRepo taskRepo) {
            this.task = task;
            this.taskRepo = taskRepo;
        }

        @Override
        public void run() {
            try {
                // 1
                Task taskInDB = taskRepo.get(this.task.getId());
                int versionLatest = taskInDB.getVersion();
                TaskStatus statusLatest = taskInDB.getStatus();
                if (!task.getStatus().equals(statusLatest))
                    return;
                if (task.getVersion() != versionLatest)
                    return;

                Map<String, Object> beans = SpringContextUtil.getBeans(TaskInvoker.class); // 找不到会抛 NoSuchBeanDefinitionException
                if (beans != null && beans.size() > 0) {
                    beans.forEach((name, bean) -> {
                        TaskInvoker taskInvoker = (TaskInvoker) bean;
                        if (taskInvoker.canInvoke(task.getType())) {
                            // 2
                            task.setStatus(TaskStatus.DOING);
                            task.setUpdateTime(LocalDateTime.now());
                            taskRepo.updateWithVersion(task);
                            task.setVersion(task.getVersion() + 1); // update 之后要重新从库中查询一次，或者直接将 version值加 1，为了后续的 updateWithVersion 操作能成功。
                            // 3
                            try {
                                boolean success = taskInvoker.invoke(task.getContent());
                                // 4
                                if (success) {
                                    // 执行成功，就将任务标记为“已完成”，否则不更新任务状态，让其停留在“执行中”，给该任务一个背再次执行的机会。
                                    task.setStatus(TaskStatus.FINISH);
                                    taskRepo.updateWithVersion(task);
                                }
                            } catch (Exception e) {
                                log.error(String.format("TaskInvoker invoke task[%s] failed.", task.getId()), e);
                                task.setStatus(TaskStatus.ERROR);
                                taskRepo.updateWithVersion(task);
                            }
                        }
                    });
                }
            } catch (Exception e) {
                log.error(String.format("Worker fail while execute task[%s].", task.getId()), e);
            }
        }
    }

    public void start() {
        init();
    }

    public void stop() {
        this.shutdown = true;
        this.workerGroup.shutdown();
    }
}
